// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: xguo, taekwon

#pragma once

#include <algorithm>
#include <functional>
#include <memory>
#include <mutex>
#include <set>
#include <string>
#include <unordered_map>
#include <utility>
#include <vector>

#include "cc/coin2/exchange/base/order/stats/stats_manager.h"
#include "coin2/base/request/i_session_pool.h"
#include "coin2/base/scheduler.h"
#include "coin2/exchange/base/api_base/fix/fix_trade_client_interface.h"
#include "coin2/exchange/base/api_base/grpc/grpc_trade_client.h"
#include "coin2/exchange/base/log/health_reporter.h"
#include "coin2/exchange/base/order/external.h"
#include "coin2/exchange/base/order/message_bus.h"
#include "coin2/exchange/base/order/order_context_manager.h"
#include "coin2/exchange/base/order/parser.h"
#include "coin2/exchange/base/order/private_ws_subscriber.h"
#include "coin2/exchange/base/order/querier.h"
#include "coin2/exchange/base/order/rate_tracker.h"
#include "coin2/exchange/base/order/recent_active_symbol.h"
#include "coin2/exchange/base/order/risk_undetected_position.h"
#include "coin2/exchange/base/order/spec_manager.h"
#include "coin2/exchange/base/order/symbol_cache.h"
#include "coin2/exchange/base/order/system_critical_error.h"
#include "coin2/exchange/base/order/update_manager.h"
#include "coin2/exchange/binance/symbology/currency.h"
#include "coin2/exchange/bitmex/symbology/currency.h"
#include "coin2/exchange/ftx/symbology/currency.h"
#include "coin2/exchange/order/feature.pb.h"
#include "coin2/exchange/prex/symbology/currency.h"
#include "naviasync/appinfo/build_info.h"

namespace coin2::exchange::base::strategy_util {
class LatencyRecorder;
class LatencyRecorderGlobal;
}  // namespace coin2::exchange::base::strategy_util

namespace coin2::exchange::base::order {

struct OrderExecution {
  using SymbolOrderMap = std::unordered_map<std::string, std::vector<const OrderContext*>>;
  using OrderSpecList = std::deque<const OrderSpec*>;
  using OrderContextList = std::deque<const OrderContext*>;

  const OrderSpec* order_spec;
  int64_t proc_order_id;
  OrderContext* oc;
  // cancel order all
  bool is_cancel_order_all = false;
  SymbolOrderMap proceed_symbol_order_map;
  SymbolOrderMap pending_symbol_order_map;
  // cancel batch orders
  bool is_cancel_batch_orders = false;
  OrderContextList cancel_proceed_order_deque;
  OrderContextList cancel_pending_order_deque;
  // place batch order
  bool is_place_batch_orders = false;
  OrderSpecList place_proceed_order_deque;
  OrderSpecList place_pending_order_deque;
};

struct LiquiditySpec {
  using LiquiditySide = coin::proto::LiquiditySide;

  LiquiditySide side;
  std::string symbol;
  double min_price;
  double max_price;
};

using GetProductHolderFunc =
    std::function<const IProductHolder*(const IProduct& product, int64_t timestamp)>;

class NativePrivateClient {
 public:
  using OrderConnectionConfig = coin2::exchange::order::OrderConnectionConfig;
  using OrderExecution = coin2::exchange::base::order::OrderExecution;
  using OrderSide = coin::proto::OrderSide;
  using OrderType = coin::proto::OrderType;
  using ContractType = coin::proto::ContractType;
  using OrderSpec = order::OrderSpec;
  using OrderContext = coin2::exchange::base::order::OrderContext;
  using IOrderContext = coin2::exchange::base::order::OrderContext;
  using IRateTracker = coin2::exchange::base::order::IRateTracker;
  using OrderLoggerCB = coin2::exchange::base::order::IRateTracker::OrderLoggerCB;
  using ActionType = coin2::exchange::base::order::IRateTracker::ActionType;
  using RateTracker = coin2::exchange::base::order::DefaultRateTracker;
  using OrderExecutionProto = coin2::exchange::base::order::OrderExecutionProto;
  using RestQueryProto = coin2::exchange::base::order::RestQueryProto;
  using OrderSubTopicId = coin::proto::OrderSubTopicId;
  using ConversionSpec = coin::proto::ConversionSpec;
  using AsioHttpContextList = std::list<std::shared_ptr<AsioHttpContext>>;

  virtual ~NativePrivateClient() = default;
  void SetSymcache(const SymbologyCache* symcache) { symcache_ = symcache; }
  void SetMessageBus(MessageBus* mb) { mb_ = mb; }
  void SetOrderLoggerCB(OrderLoggerCB callback) { order_logger_cb_ = callback; }
  void SetGetProductHolder(GetProductHolderFunc func) { get_product_holder_ = func; }

  int GetSize() const { return auths_.GetSize(); }
  BaseRestAuth* GetNth(int n) const { return auths_.GetNth(n); }

  BaseRestAuth* GetFirstAuth() const { return auths_.GetFirst(); }
  virtual BaseRestAuth* GetAuth() const { return auths_.GetRandom(); }
  virtual void Prepare() {}

  virtual std::shared_ptr<AsioHttpContext> SubmitOrderExecution(const OrderExecutionProto& order) {
    return nullptr;
  }
  virtual std::shared_ptr<AsioHttpContext> SubmitRestQuery(const RestQueryProto& query) {
    return nullptr;
  }

  virtual AsioHttpContextList SubmitRestQueryList(const RestQueryProto& query) { return {}; }

  virtual std::shared_ptr<AsioHttpContext> PlaceOrderNorm(const OrderExecution&) = 0;
  virtual std::shared_ptr<AsioHttpContext> CancelOrderNorm(const OrderExecution&) = 0;
  virtual std::shared_ptr<AsioHttpContext> AmendOrderNorm(const OrderExecution&) {
    CHECK(false) << "AmendOrderNorm is not implemented";
    return {};
  }
  virtual std::shared_ptr<AsioHttpContext> SubmitConversion(const ConversionSpec&) {
    CHECK(false) << "SubmitConversion is not implemented";
    return {};
  }

  // rate tracker related
  IRateTracker* rate_tracker() { return rate_tracker_.get(); }
  bool HasEnoughRateLimit(int64_t ts, ActionType action_type) const {
    if (!rate_tracker_) {
      return true;
    }
    auto ret = rate_tracker_->HasEnoughRateLimit(ts, action_type);
    // write order log
    if (!ret && order_logger_cb_) {
      rate_tracker_->WriteRateLimitReportMessage(order_logger_cb_);
    }
    return ret;
  }
  void Push(int32_t num_reqs, int64_t timestamp, ActionType action_type) {
    if (!rate_tracker_) {
      return;
    }
    rate_tracker_->Push(num_reqs, timestamp, action_type);
  }
  const base::symbology::IProductHolder* GetProductHolder(const IProduct& product, int64_t ts) {
    return get_product_holder_(product, ts);
  }

 protected:
  RoundRobinRestAuth auths_;
  const SymbologyCache* symcache_;
  MessageBus* mb_;
  MarketExchangeApi mea_;
  GetProductHolderFunc get_product_holder_;

  std::unique_ptr<IRateTracker> rate_tracker_ = {};
  OrderLoggerCB order_logger_cb_;
};

class RestSubscriber;

class BaseOrderProcessor {
 public:
  using OrderSubTopicId = base::order::OrderSubTopicId;
  using RawMessage = BaseParser::RawMessage;
  using RiskEvent = coin::proto::RiskEvent;
  using RecentActiveSymbol = coin2::exchange::base::order::RecentActiveSymbol;
  using NftTokenInfo = ::coin::proto::NftTokenInfo;
  using FundingFeeBundle = ::coin::proto::FundingFeeBundle;

 public:
  // temporary workaround for sim OG
  BaseOrderProcessor() : order_upd_mgr_("default", MarketExchangeApi()) {}

  BaseOrderProcessor(
      const std::string& system_name,
      const MarketExchangeApi& mea,
      IOrderSubscriber* callback,
      order_util::OrderLogger* logger,
      order_util::OrderLogger* order_packet_logger,
      bool enable_raw_og_log)
      : order_upd_mgr_(system_name, mea),
        callback_(callback),
        logger_(logger),
        order_packet_logger_(order_packet_logger),
        enable_raw_og_log_(enable_raw_og_log) {
    mutable_upd_mgr()->GetAccountInfoMutable()->set_owner(logger_->request().owner());
    mea_ = mea;
    order_rate_tracker_ = std::make_unique<OrderRateTracker>(mea);
    risk_undetected_position_ = std::make_unique<RiskUndetectedPosition>();
    risk_undetected_position_->RegisterRiskEventHandler(
        std::bind(&BaseOrderProcessor::HandleRiskEvent, this, std::placeholders::_1));

    // 16777213 is a prime number that is smaller than 2**24, so msg_id high
    og_id_ = reinterpret_cast<std::uintptr_t>(this) % 16777213UL;
    msg_id_ = (og_id_ << 32);
    LOG(INFO) << "og_id: " << og_id_ << ", "
              << "msg_id: " << msg_id_;
    auto build_info = GetBuildInfo();
    SPDLOG_INFO("git revision: {}", build_info.revision());
  }

  int64_t og_id() const { return og_id_; }
  virtual const OrderUpdateManager& upd_mgr() const { return order_upd_mgr_; }
  virtual OrderUpdateManager* mutable_upd_mgr() { return &order_upd_mgr_; }
  OrderRateTracker* order_rate_tracker() { return order_rate_tracker_.get(); }

  void SetSymcache(const SymbologyCache* symcache) {
    symcache_ = symcache;
    for (const auto* prod : symcache->GetProducts()) {
      order_upd_mgr_.AddProduct(prod);
    }
  }
  void SetMessageBus(MessageBus* mb) { mb_ = mb; }
  void SetOcm(OrderContextManager* ocm) { ocm_ = ocm; }
  void SetParser(BaseParser* parser) { parser_ = parser; }
  void SetRestSubscriber(RestSubscriber* rest_sub) { rest_subscriber_ = rest_sub; }

  BaseParser* parser() {
    CHECK(parser_);
    return parser_;
  }

  OrderContextManager* order_manager() {
    CHECK(ocm_);
    return ocm_;
  }

  void Prepare() {}
  void PublishEvent(const std::string& channel, void* data) { mb_->PublishEvent(channel, data); }

  void AddNativeSymbol(const std::string& native_symbol, IOrderSubscriber* subscriber) {
    mutable_upd_mgr()->AddNativeSymbol(native_symbol, subscriber);
    auto product = symcache_->GetProductByNative(native_symbol);
    auto* ptr = mutable_upd_mgr()->GetAccountInfoMutable();
    ptr->AddProduct(product->absolute_norm());
    auto* pi = symcache_->GetProductInfoByNorm(product->absolute_norm());

    auto mea = upd_mgr().mea();
    if (symcache_->is_futures() && symcache_->IsDeprecatedExchange(*product, mea)) {
      if (pi->is_inverse()) {
        ptr->AddCurrency(product->base().currency());
      } else {
        ptr->AddCurrency(product->quote().currency());
      }
    } else {
      ptr->AddCurrency(product->base().currency());
      ptr->AddCurrency(product->quote().currency());
    }

    // Temporary workaround, will change later.
    if (!is_extra_symbol_initialized) {
      add_extra_symbol(mea.exchange);
      is_extra_symbol_initialized = true;
    }
  }

  void add_extra_symbol(const ExchangeType& exchange_type) {
    auto* ptr = mutable_upd_mgr()->GetAccountInfoMutable();
    if (exchange_type == ExchangeType::Ftx && symcache_->is_futures()) {
      using coin2::exchange::ftx::symbology::FtxCurrency;
      auto timestamp = GetCurrentTimestamp();
      const static std::vector<std::string> currency_list = {
          "BTC",
          "ETH",
          "USDT",
          "FTT",
          "RAY",
          "SRM",
          "ETHW",
          "XRP",
          "DOGE",
          "CUSDT",
          "WUSDT",
          "GBTC",
          "WBTC",
          "XRP",
          "YFI",
          "UNI",
          "TRX"};
      for (const auto& currency_str : currency_list) {
        if (auto currency = FtxCurrency::FromNormStrNoThrow(currency_str, timestamp)) {
          ptr->AddCurrency(currency->currency());
        }
      }
    }
    if (exchange_type == ExchangeType::Binance) {
      using coin2::exchange::binance::symbology::BinanceCurrency;
      auto timestamp = GetCurrentTimestamp();
      if (auto currency = BinanceCurrency::FromNormStrNoThrow("BNB", timestamp)) {
        ptr->AddCurrency(currency->currency());
      }
      if (auto currency = BinanceCurrency::FromNormStrNoThrow("BUSD", timestamp)) {
        ptr->AddCurrency(currency->currency());
      }
      if (auto currency = BinanceCurrency::FromNormStrNoThrow("USDC", timestamp)) {
        ptr->AddCurrency(currency->currency());
      }
      if (auto currency = BinanceCurrency::FromNormStrNoThrow("USDT", timestamp)) {
        ptr->AddCurrency(currency->currency());
      }
    }
    if (exchange_type == ExchangeType::Bitmex && symcache_->is_futures()) {
      using coin2::exchange::bitmex::symbology::BitmexCurrency;
      auto timestamp = GetCurrentTimestamp();
      if (auto currency = BitmexCurrency::FromNormStrNoThrow("BTC", timestamp)) {
        ptr->AddCurrency(currency->currency());
      }
    }
    if (exchange_type == ExchangeType::Prex) {
      using coin2::exchange::prex::symbology::PrexCurrency;
      auto timestamp = GetCurrentTimestamp();
      if (auto currency = PrexCurrency::FromNormStrNoThrow("USD", timestamp)) {
        ptr->AddCurrency(currency->currency());
      }
    }
  }

  void UpdateAccountBalance(const AccountBalance& acc_balance) {
    mutable_upd_mgr()->UpdateAccountBalance(acc_balance, current_ts_);
    auto account_balance = upd_mgr().account_info().GetAccountBalance();
    account_balance.set_query_ts(current_ts_);  // TODO: should this be moved to parser ?
    logger_->WriteBalance(account_balance, true, current_ts_);
  }

  void ProcessUndetectPosition(
      ProductPosition* position,
      std::vector<ProductPosition>* position_not_in_symcache);

  std::vector<ProductPosition> FilterAccountPosition(AccountPosition* acc_position);

  void HandleUndetectedPosition(std::vector<ProductPosition>&& undetected_position);

  void UpdateAccountPosition(const AccountPosition& acc_position) {
    mutable_upd_mgr()->UpdateAccountPosition(acc_position, current_ts_);
    auto account_position = upd_mgr().account_info().GetAccountPosition();
    account_position.set_query_ts(current_ts_);  // TODO: should this be moved to parser ?
    logger_->WritePosition(account_position, true, current_ts_);
  }

  void UpdateCurrencyTransfer(const CurrencyTransferBundle& ccy_transfer) {
    mutable_upd_mgr()->UpdateCurrencyTransfer(ccy_transfer, current_ts_);
    logger_->WriteTransfer(upd_mgr().account_info().GetCurrencyTransfer(), true, current_ts_);
  }

  void UpdateFundingFee(const FundingFeeBundle& funding_fee) {
    mutable_upd_mgr()->UpdateFundingFee(funding_fee, current_ts_);
    logger_->WriteFundingFee(upd_mgr().account_info().GetFundingFee(), true, current_ts_);
  }

  void UpdateNftTokenInfo(const NftTokenInfo& nft_token_info) {
    mutable_upd_mgr()->UpdateNftTokenInfo(nft_token_info, current_ts_);
    logger_->WriteNftTokenInfo(upd_mgr().account_info().GetNftTokenInfo(), true, current_ts_);
  }

  void HandleOrderEvent(const OrderEvent& order_event);
  void ProcessRawMessage(const RawMessage& msg);
  void PreProcessMessage(const RawMessage& msg);
  void UpdateTimestamp(int64_t ts) {
    if (ts >= current_ts_) {
      current_ts_ = ts;
    }
  }
  int64_t get_ts() const {
    CHECK_GT(current_ts_, 10000);
    return current_ts_;
  }

  void set_store_parsed_message(bool set) { store_parsed_message_ = set; }

  void SetRawOGLog(bool is_enable_raw_og_log) { enable_raw_og_log_ = is_enable_raw_og_log; }

  StatsInfo GetStatsInfo() {
    StatsInfo info = info_;
    info_.Reset();
    return info;
  }

  std::pair<StatsInfo, int> GetOGInfo() { return std::make_pair(info_, ws_reset_num_); }

  int GetWsResetNum() {
    int ws_reset_num = ws_reset_num_;
    ws_reset_num_ = 0;
    return ws_reset_num;
  }

  void InitRecentActiveSymbol() {
    std::call_once(recent_active_symbol_flag, [this]() {
      recent_active_symbol_ = std::make_unique<RecentActiveSymbol>(300'000'000'000LL);
    });
  }

  bool HasActiveEvent(const std::string& native_symbol) {
    if (!recent_active_symbol_) {
      return true;
    }
    auto ts = current_ts_ > 0 ? current_ts_ : GetCurrentTimestamp();
    return recent_active_symbol_->HasActiveEvent(ts, native_symbol);
  }

  std::pair<int64_t, std::string> GetRawOrderData() { return raw_order_data_; }

  void HandleHttpError(const std::shared_ptr<AsioHttpContext>&);
  void HandleRiskEvent(const RiskEvent& risk_event);

 public:
  coin2::exchange::base::strategy_util::LatencyRecorder* latency_recorder_{nullptr};

 protected:
  std::unique_ptr<OrderRateTracker> order_rate_tracker_;
  std::unique_ptr<RiskUndetectedPosition> risk_undetected_position_{};
  std::unordered_map<std::string, std::unique_ptr<IProduct>> native_symbol_to_product_map_;
  OrderUpdateManager order_upd_mgr_;
  IOrderSubscriber* callback_;
  order_util::OrderLogger* logger_;
  order_util::OrderLogger* order_packet_logger_;
  OrderContextManager* ocm_;
  MarketExchangeApi mea_;

  const SymbologyCache* symcache_;
  MessageBus* mb_{nullptr};
  BaseParser* parser_;
  int64_t msg_id_{0};
  int64_t og_id_{0};
  int64_t current_ts_{0};
  bool store_parsed_message_{false};
  RestSubscriber* rest_subscriber_{0};
  bool enable_raw_og_log_;

  // stats info
  StatsInfo info_;
  // ws reset
  int ws_reset_num_{0};
  // raw message snapshot
  std::pair<int64_t, std::string> raw_order_data_;
  bool is_extra_symbol_initialized = false;

  int64_t last_check_timeout_order_ts_ = 0;
  int64_t last_check_zombie_order_ts_ = 0;

  std::once_flag recent_active_symbol_flag;
  std::unique_ptr<RecentActiveSymbol> recent_active_symbol_ = {};
};

class RestSubscriber {
 public:
  using PrivateSubscriberConfig = coin2::exchange::order::PrivateSubscriberConfig;
  using OrderSubTopicId = coin2::exchange::base::order::OrderSubTopicId;
  using ActionType = NativePrivateClient::ActionType;
  using GrpcTradeClient = coin2::exchange::base::api_base::grpc::GrpcTradeClient;

 public:
  static constexpr int64_t kOneSecToNs = 1'000'000'000LL;

 public:
  RestSubscriber(
      NativePrivateClient* rest_client,
      base::api_base::NativeHttpAsioDriver* http_driver,
      const PrivateSubscriberConfig& config)
      : config_(config), http_driver_(http_driver), rest_client_(rest_client) {}
  virtual ~RestSubscriber() = default;

  virtual void QueryTransfer() {}
  virtual void RegisterPubTopic() {}
  virtual void SubscribeAsync() {}
  virtual void StartAsync() {
    // TODO(xguo) Read from config
    if (symcache_->mea().exchange == ExchangeType::Bitmex) {
      scheduler_->SetCheckPeriod(2'000'000'000LL);  // Bitmex
    }
    if (symcache_->mea().exchange == ExchangeType::Bitstamp) {
      scheduler_->SetCheckPeriod(500'000'000LL);  // Bitstamp
    }
    if (symcache_->mea().exchange == ExchangeType::Gdax) {
      scheduler_->SetCheckPeriod(500'000'000LL);  // Gdax
    }

    SubscribeAsync();
    scheduler_->Start();
  }
  virtual void StopAsync() { scheduler_->Stop(); }

  void HandleQuery(const std::shared_ptr<AsioHttpContext>& http_context);
  auto GetConfig() const -> const PrivateSubscriberConfig& { return config_; }
  void SetIsReadonlyOrderSystem(bool is_readonly) { is_readonly_order_system_ = is_readonly; }
  void SetProcessor(BaseOrderProcessor* processor) { processor_ = processor; }
  void SetFlowSub(base::api_base::LiveAsyncExecutor* flow_sub) {
    scheduler_.reset(new Scheduler(flow_sub->MainIoc()));
  }
  void SetMessageBus(MessageBus* mb) { mb_ = mb; }
  void SetGrpcClient(GrpcTradeClient* grpc_client) { grpc_client_ = grpc_client; }
  void SetSymcache(const SymbologyCache* symcache) { symcache_ = symcache; }
  bool is_swap() const {
    CHECK(symcache_);
    return symcache_->is_swap();
  }
  std::set<std::string> native_symbols() {
    std::set<std::string> res;
    for (const auto* product : symcache_->GetProducts()) {
      res.emplace(product->order_native_symbol());
    }
    return res;
  }

 protected:
  bool SkipQuery(const std::string& task_name, const std::string& native_symbol);

 protected:
  typedef std::function<void()> TaskFuncType;
  typedef std::function<bool()> SkipFuncType;

  struct TaskWrapper {
    TaskFuncType task_func;
    SkipFuncType skip_func;

    void operator()() {
      if (skip_func()) {
        return;
      }
      task_func();
    }
  };

  // add single task
  void AddPeriodicTask(
      TaskFuncType task_func,
      double period_sec,
      const std::string& task_name,
      double delay_sec = 0) {
    SYSTEM_CRITICAL_ERROR(period_sec < kOneSecToNs, "invalid period second of task, please check");
    SYSTEM_CRITICAL_ERROR(delay_sec < kOneSecToNs, "invalid delay second of task, please check");
    scheduler_
        ->AddPeriodicTask(task_func, period_sec * kOneSecToNs, task_name, delay_sec * kOneSecToNs);
  }

  // add mutliple tasks based on container list
  // e.g.  AddPeriodicTask(func, interval_sec, task_name, currency_list, symbol_list, ...)
  template <
      typename TaskFunc,
      typename SkipFunc,
      typename... Args,
      typename = std::enable_if_t<std::is_invocable_r_v<
          void,
          TaskFunc,
          typename std::remove_reference_t<Args>::value_type...>>,
      typename = std::enable_if_t<std::is_invocable_r_v<
          bool,
          SkipFunc,
          typename std::remove_reference_t<Args>::value_type...>>>
  void AddPeriodicTask(
      TaskFunc task_func,
      SkipFunc skip_func,
      double period_sec,
      const std::string& task_name,
      Args&&... args) {
    SYSTEM_CRITICAL_ERROR(period_sec < kOneSecToNs, "invalid period second of task, please check");
    processor_->InitRecentActiveSymbol();
    auto add_func = [this,
                     period_sec,
                     &task_name](TaskFuncType task_func, SkipFuncType skip_func, double delay_sec) {
      TaskWrapper task;
      task.skip_func = skip_func;
      task.task_func = task_func;
      AddPeriodicTask(task, period_sec, task_name, delay_sec);
    };

    AddPeriodicTaskImpl(task_func, skip_func, add_func, 0, period_sec, std::forward<Args>(args)...);
  }

  template <
      typename TaskFunc,
      typename... Args,
      typename = std::enable_if_t<std::is_invocable_r_v<
          void,
          TaskFunc,
          typename std::remove_reference_t<Args>::value_type...>>>
  void AddPeriodicTask(
      TaskFunc task_func,
      double period_sec,
      const std::string& task_name,
      Args&&... args) {
    SYSTEM_CRITICAL_ERROR(period_sec < kOneSecToNs, "invalid period second of task, please check");
    auto add_func = [this,
                     period_sec,
                     &task_name](TaskFuncType task_func, SkipFuncType skip_func, double delay_sec) {
      TaskWrapper task;
      task.skip_func = skip_func;
      task.task_func = task_func;
      AddPeriodicTask(task, period_sec, task_name, delay_sec);
    };
    auto skip_func = [](auto... args) { return false; };
    AddPeriodicTaskImpl(task_func, skip_func, add_func, 0, period_sec, std::forward<Args>(args)...);
  }

 private:
  template <
      typename TaskFunc,
      typename SkipFunc,
      typename C,
      typename... Args,
      typename = std::enable_if_t<std::is_invocable_r_v<
          void,
          TaskFunc,
          typename C::value_type,
          typename std::remove_reference_t<Args>::value_type...>>,
      typename = std::enable_if_t<std::is_invocable_r_v<
          bool,
          SkipFunc,
          typename C::value_type,
          typename std::remove_reference_t<Args>::value_type...>>>
  void AddPeriodicTaskImpl(
      TaskFunc task_func,
      SkipFunc skip_func,
      std::function<void(TaskFuncType, SkipFuncType, double)> add_func,
      double delay_start_sec,
      double delay_len_sec,
      const C& container,
      Args&&... args) {
    delay_len_sec /= container.size();

    int index = 0;
    for (const auto& item : container) {
      AddPeriodicTaskImpl(
          [item, task_func](auto&&... args) {
            task_func(item, std::forward<decltype(args)>(args)...);
          },
          [item, skip_func](auto&&... args) -> bool {
            return skip_func(item, std::forward<decltype(args)>(args)...);
          },
          add_func,
          delay_start_sec + index * delay_len_sec,
          delay_len_sec,
          std::forward<Args>(args)...);
      ++index;
    }
  }

  template <
      typename TaskFunc,
      typename SkipFunc,
      typename = std::enable_if_t<std::is_invocable_r_v<void, TaskFunc>>,
      typename = std::enable_if_t<std::is_invocable_r_v<bool, SkipFunc>>>
  void AddPeriodicTaskImpl(
      TaskFunc task_func,
      SkipFunc skip_func,
      std::function<void(TaskFuncType, SkipFuncType, double)> add_func,
      double delay_start_sec,
      double delay_len_sec) {
    add_func(task_func, skip_func, delay_start_sec);
  }

 protected:
  const PrivateSubscriberConfig config_;
  base::api_base::NativeHttpAsioDriver* http_driver_;
  NativePrivateClient* rest_client_;
  GrpcTradeClient* grpc_client_ = nullptr;

  std::string topic_name_;
  executor::LiveAsyncExecutor* flow_sub_;
  BaseOrderProcessor* processor_;
  const SymbologyCache* symcache_;
  BaseParser::RawMessage raw_msg_;
  MessageBus* mb_;
  bool is_readonly_order_system_ = false;

 private:
  std::shared_ptr<Scheduler> scheduler_;
};

class WsSubscriber : public base::api_base::BaseRawAsyncSubscriber {
 public:
  using BaseRawAsyncSubscriber::BaseRawAsyncSubscriber;
  using OrderConnectionConfig = coin2::exchange::order::OrderConnectionConfig;

  WsSubscriber(
      const MarketExchangeApi& mea,
      base::api_base::LiveAsyncExecutor* flow_sub,
      const OrderConnectionConfig& conn_config);

  void RegisterPubTopic() override { flow_sub_->distributor()->Publish(spec_->topic); }
  void SubscribeAsync() override;

  void SetProcessor(BaseOrderProcessor* processor) { processor_ = processor; }
  void SetMessageBus(MessageBus* mb) { mb_ = mb; }

  void SetListenKey(const std::string& channel, void* data) {
    if (mea_.exchange == ExchangeType::Lbank) {
      std::pair<bool, std::string> pair(*static_cast<std::pair<bool, std::string>*>(data));
      if (pair.first) {
        return;
      }

      if (listen_key_.empty()) {
        listen_key_ = pair.second;
        conn_->SetListenKey(listen_key_);
        LOG(INFO) << "set listen key: " << channel << ", " << listen_key_;
        return;
      }

      LOG(WARNING) << "[Lbank] requires reconnect when listen key changed, prev:" << listen_key_
                   << " new:" << pair.second;
      listen_key_ = pair.second;
      conn_->SetListenKey(listen_key_);
      conn_->ForceResetConnection();
      return;
    }
    listen_key_ = *(static_cast<std::string*>(data));
    conn_->SetListenKey(listen_key_);
    LOG(INFO) << "set listen key: " << channel << ", " << listen_key_;
  }

  void SetUserId(const std::string& channel, void* data) {
    user_id_ = *(static_cast<std::string*>(data));
    conn_->SetUserId(user_id_);
    LOG(INFO) << "set user id: " << channel << ", " << user_id_;
  }

  void SetBullet(const std::string& channel, void* data) {
    bullet_info_ = *(static_cast<std::vector<std::string>*>(data));
    conn_->SetBullet(bullet_info_);
    LOG(INFO) << "set bullet: " << channel;
  }

  void SetLastRecvMsgTs(const std::string& channel, void* data) {
    int64_t timestamp = *(static_cast<int64_t*>(data));
    if (0 == timestamp) {
      timestamp = GetCurrentTimestamp();
    }
    conn_->SetLastRecvMsgTs(timestamp);
  }

  void ForceResetConnection(const std::string& channel, void* data) {
    conn_->ForceResetConnection();
  }

  void SetIsReadonlyOrderSystem(bool is_readonly) { is_readonly_order_system_ = is_readonly; }

  bool CheckBlacklistedMea(const MarketExchangeApi& mea) { return config_.ws_disable_subscribe(); }

 protected:
  const MarketExchangeApi mea_;
  OrderConnectionConfig config_;
  BaseOrderProcessor* processor_;
  MessageBus* mb_;
  std::unique_ptr<PrivateWsSubscriberSpec> spec_;
  std::shared_ptr<BasePrivateWsConnection> conn_;
  bool is_readonly_order_system_ = false;

  // private ws connection requires fields
  std::string listen_key_;                // for Binance|kraken|Lbank|Bit|Mexc|Apollox
  std::string user_id_;                   // for Gateio
  std::vector<std::string> bullet_info_;  // for Kucoin
};

class BaseOrderGateway {
 public:
  using IProduct = symbology::IProduct;
  using OrderSpec = order::OrderSpec;
  using BatchOrdersSpec = order::BatchOrdersSpec;
  using BatchCancelSpec = order::BatchCancelSpec;
  using OrderContext = order::OrderContext;
  using OrderContextManager = order::OrderContextManager;
  using OrderGatewayConfig = coin::proto::OrderGatewayConfig;
  using ExchangeType = coin::proto::ExchangeType;
  using MarketType = coin::proto::MarketType;
  using IFixTradeClient = coin2::exchange::base::api_base::fix::IFixTradeClient;
  using ActionType = NativePrivateClient::ActionType;
  using LatencyRecorder = coin2::exchange::base::strategy_util::LatencyRecorder;
  using LatencyRecorderGlobal = coin2::exchange::base::strategy_util::LatencyRecorderGlobal;
  using GrpcTradeClient = coin2::exchange::base::api_base::grpc::GrpcTradeClient;
  using GrpcContext = coin2::exchange::base::api_base::grpc::GrpcContext;

 public:
  // workaround for sim OG.
  BaseOrderGateway();

  BaseOrderGateway(
      const MarketExchangeApi& mea,
      const OrderGatewayConfig& og_config,
      executor::LiveAsyncExecutor* exec,
      NativePrivateClient* rest_client,
      base::api_base::NativeHttpAsioDriver* http_driver,
      order_util::OrderLogger* order_logger);

  // !!!Remove later!!!
  virtual const OrderContextManager& order_manager() const { return *ocm_; }
  virtual OrderContextManager* mutable_order_manager() { return ocm_; }

  // Simulation specific
  virtual void UpdateTimeForSim(int64_t ts) {}
  virtual void SetApiRateLimitCheckForSim(bool check) {}

  virtual std::optional<int64_t> TryRegisterForSim(const IProduct& product) {
    return std::optional<int64_t>();
  }

  virtual std::optional<int64_t> SubmitOrderForSim(int64_t account_id, const OrderSpec& spec) {
    return std::optional<int64_t>();
  }

  // TODO(xguo) move to parser.
  virtual void GetRateLimitInfo(const std::shared_ptr<AsioHttpContext>& http_context);

  // move to OrderSubsystem.
  bool need_exchange_id_to_cancel() const { return need_exchange_id_to_cancel_; }
  virtual bool is_supporting_ioc() const { return is_supporting_ioc_; }

  // setters
  void SetSymcache(const SymbologyCache* symcache) { symcache_ = symcache; }
  void SetMessageBus(MessageBus* mb) { mb_ = mb; }
  void SetProcessor(BaseOrderProcessor* processor) { processor_ = processor; }
  virtual IRateTracker* GetMutableRateTracker() { return rest_client_->rate_tracker(); }
  void SetOcm(OrderContextManager* ocm) { ocm_ = ocm; }
  void SetFixClient(IFixTradeClient* fix_client) { fix_client_ = fix_client; }
  void SetGrpcClient(GrpcTradeClient* grpc_client) { grpc_client_ = grpc_client; }

  virtual void set_latency_recorder(
      coin2::exchange::base::strategy_util::LatencyRecorder* latency_recorder);

  // for latency profile debugging
  auto GetCurrContextIdWithNoRegisteredOG() -> int64_t;
  void RemoveContextIdFromRegisteredOGSet(int64_t context_id);

  void Subscribe(const std::string& channel, const MessageBus::Callback& cb) {
    mb_->Subscribe(channel, cb);
  }

  void PublishEvent(const std::string& channel, void* data) { mb_->PublishEvent(channel, data); }
  bool IsAcceptableValue(const OrderSpec& order_spec) const;
  NativePrivateClient* rest_client() { return rest_client_; }
  void Prepare() {}

  virtual void HandleResponse(
      std::vector<const OrderContext*> oc_list,
      std::shared_ptr<AsioHttpContext> http_context);

  virtual void HandleGrpcResponse(std::shared_ptr<GrpcContext> context);

  // Interface
  virtual auto SubmitOrder(const OrderSpec& order_spec) -> std::optional<int64_t>;
  virtual bool CancelOrder(int64_t proc_order_id);
  virtual int CancelBatchOrders(const BatchCancelSpec& batch_cancel_spec);
  virtual bool CancelOrderAll(const std::string& symbol);
  virtual auto SubmitBatchOrders(const BatchOrdersSpec& batch_orders_spec)
      -> int /* submitted_cnt */;
  virtual auto AmendOrder(int64_t proc_order_id, double amend_price, double amend_qty)
      -> std::optional<int64_t>;
  virtual bool SubmitConversion(const ConversionSpec& conversion_spec);
  virtual bool SubmitLiquidity(const LiquiditySpec& liquidity_spec);

  bool IsSupportMultiSubmitOrder() const { return is_supporting_multi_submit_order_; }
  bool IsSupportMultiCancelOrder() const { return is_supporting_multi_cancel_order_; }

 protected:
  // cancel order all helper
  bool IsCancelOrderAllByAccount(std::string_view symbol) const {
    return "all" == symbol || "ALL" == symbol;
  }
  void PrepareCancelOrdersByProduct(
      int64_t timestamp,
      const std::string& product,
      OrderExecution::SymbolOrderMap* pending_symbol_order_map);
  bool PrepareCancelAllOrderExecution(
      int64_t timestamp,
      const std::string& product,
      OrderExecution* order_exe);
  void PrepareSubmitBatchOrdersExecution(
      const std::deque<const OrderSpec*>& order_spec_deque,
      OrderExecution* order_exe);
  void PrepareCancelBatchOrdersExecution(
      const std::deque<const OrderContext*>& order_context_deque,
      OrderExecution* order_exe);

  void PrepareSubmitOrderExecutionProto(const OrderSpec& order_spec);
  void PrepareCancelOrderExecutionProto(int64_t proc_order_id);
  void PrepareCancelBatchExecutionProto(const std::vector<int64_t>& proc_order_ids);
  void PrepareAmendOrderExecutionProto(
      int64_t proc_order_id,
      int64_t orig_proc_order_id,
      double amend_price,
      double amend_qty,
      const OrderSpec& amend_order_spec);
  void PrepareCancelAllExecutionProto(const std::string& native_symbol);
  void PrepareSubmitBatchExecutionProto(const std::deque<OrderSpec>& order_spec_deque);
  void PrepareSubmitBatchExecutionProto(const std::deque<const OrderSpec*>& order_spec_deque);
  void SetOrderExecutionError(const std::string& err_msg);
  void SetOrderExecutionError(const std::string& err_msg, int error_code);

  void ClearResult() {
    result_.raw_msg_id = 0;
    result_.order_execution.Clear();
    result_.type = coin::proto::OrderSubTopicId::ORDER_EXECUTION;
  }
  void InitWithMea(const MarketExchangeApi& mea);

 protected:
  const OrderGatewayConfig og_config_;
  OrderContextManager* ocm_;

  executor::LiveAsyncExecutor* flow_sub_;
  order_util::OrderLogger* order_logger_;
  coin2::exchange::base::strategy_util::LatencyRecorder* latency_recorder_{nullptr};
  std::unordered_set<int64_t> og_registered_context_id_set_;  // for latency profile debugging

  const SymbologyCache* symcache_{nullptr};
  MessageBus* mb_{nullptr};

  base::api_base::NativeHttpAsioDriver* http_driver_{nullptr};
  BaseOrderProcessor* processor_;
  NativePrivateClient* rest_client_;
  GrpcTradeClient* grpc_client_ = nullptr;
  IFixTradeClient* fix_client_;
  const MarketExchangeApi mea_;
  bool is_swap_;
  bool is_futures_;
  bool need_exchange_id_to_cancel_;
  bool is_supporting_ioc_;
  bool is_supporting_post_only_;
  bool is_supporting_multi_cancel_order_;
  bool is_supporting_multi_submit_order_;
  BaseParser::RawMessage result_;
};

class BaseOrderSubsystem : public IAsyncActor {
 public:
  using AccountRequestProto = coin::proto::AccountRequestProto;
  using BaseOrderGateway = order::BaseOrderGateway;
  using BaseOrderProcessor = order::BaseOrderProcessor;
  using ExchangeType = coin::proto::ExchangeType;
  using IFixTradeClient = coin2::exchange::base::api_base::fix::IFixTradeClient;
  using GrpcTradeClient = coin2::exchange::base::api_base::grpc::GrpcTradeClient;
  using MarketType = coin::proto::MarketType;
  using OrderLogger = coin2::exchange::base::order_util::OrderLogger;
  using OrderSubsystemFeature = coin2::exchange::order::OrderSubsystemFeature;
  using OrderSystemConfig = coin2::exchange::order::OrderSystemConfig;
  using PrivateSubscriberConfig = coin2::exchange::order::PrivateSubscriberConfig;
  using RttInfo = coin2::base::request::RttInfo;
  using RequestStatus = coin2::base::request::RequestStatus;

 public:
  BaseOrderSubsystem(
      const std::string& system_name,
      const OrderSystemConfig& oss_config,
      IOrderSubscriber* callback)
      : callback_(callback),
        system_name_(system_name),
        mea_(MarketExchangeApi::FromString(oss_config.mea())) {}

  BaseOrderSubsystem(
      const std::string& system_name,
      const MarketExchangeApi& mea,
      const di::ConstantInjection& ci,
      const OrderSystemConfig& oss_config,
      executor::LiveAsyncExecutor* exec,
      IOrderSubscriber* callback,
      GetProductHolderFunc get_product_holder);

  virtual ~BaseOrderSubsystem();

  // Get functional module in the OG subsystem.
  base::api_base::NativeHttpAsioDriver* http_driver() { return http_driver_.get(); }
  virtual BaseOrderGateway* gateway() { return gateway_.get(); }
  virtual BaseOrderProcessor& processor() { return *(processor_.get()); }

  virtual bool is_supporting_separate_pos() const;
  virtual bool is_sim_os() const { return false; }
  virtual bool is_ready() const;
  const auto products() const { return products_; }
  const std::string& system_name() const { return system_name_; }
  const MarketExchangeApi& mea() const { return mea_; }
  OrderSubsystemFeature OsFeature() const { return os_feature_; }
  const AccountRequestProto& GetAccountRequest() const { return account_request_; }

  OrderContextManager* order_manager() {
    CHECK(ocm_);
    return ocm_.get();
  }

  virtual OrderQuerier* GenOrderQuerierForSim(int64_t) { return nullptr; }
  OrderQuerier* GenOrderQuerier() /*const*/ {
    // FIX: this leaks memory.
    return new OrderQuerier(processor().upd_mgr(), *ocm_, is_supporting_separate_pos());
  }

  void StartAsync();
  void StopAsync();
  void LiveInit(const OrderSystemConfig& oss_config, executor::LiveAsyncExecutor* exec);

  void SetParser(std::unique_ptr<BaseParser>&& parser) { parser_ = std::move(parser); }
  void SetRestSubscriber(std::unique_ptr<RestSubscriber>&& rest_subscriber) {
    rest_subscriber_ = std::move(rest_subscriber);
  }
  void SetRestClient(std::unique_ptr<NativePrivateClient>&& rest_client) {
    rest_client_ = std::move(rest_client);
  }

  auto GetProductHolder(const IProduct& product, int64_t ts)
      -> const base::symbology::IProductHolder* {
    return get_product_holder_(product, ts);
  }
  void GetHealthReport(int indent, std::string* report) const {
    // [HealthReport]
    //     [Order]
    //         - Huobi (OK)
    //         - Okex (STARTING, uptime=23h)
    //               a1 (..)

    for (int i = 0; i < indent * 4; i++) {
      *report += ' ';
    }
    *report += system_name();
    *report += " (?)\n";
  }

  void SetRawOGLog(bool is_enable_raw_og_log) { processor_->SetRawOGLog(is_enable_raw_og_log); }

  void RegisterHealthReporter(std::function<void(RequestStatus)> reporter_func) {
    reporter_func_ = reporter_func;
  }

  StatsInfo GetStatsInfo() { return processor_->GetStatsInfo(); }

  int GetWsResetNum() { return processor_->GetWsResetNum(); }

  RttInfo GetRttInfo() { return http_driver_->GetRttInfo(); }

  RequestStatus GetRequestInfo() { return http_driver_->GetRequestInfo(); }
  void GetApiEndPointStats(RequestStatus info);

  // Exported methods
  std::map<std::string, std::function<std::string(const std::string&)>> ExportMethods();
  std::string GetOssFeatures(const std::string& req);
  std::string GetOssConfig(const std::string& req);
  std::string GetOpenOrders(const std::string& req);
  std::string SetRawOGLog(const std::string& req);
  std::string GetOGInfo(const std::string&);
  std::string GetGitHash(const std::string&);
  std::string GetRawOrderData(const std::string&);

 protected:
  void SetOsFeature(const MarketExchangeApi& mea);

 protected:
  IOrderSubscriber* callback_;
  const std::string system_name_;
  const MarketExchangeApi mea_;
  std::vector<const IProduct*> products_;
  SymbologyCache symcache_;
  OrderSubsystemFeature os_feature_;
  AccountRequestProto account_request_;

  const OrderSystemConfig oss_config_;
  LiveAsyncExecutor* live_executor_;

  std::unique_ptr<MessageBus> message_bus_;
  std::unique_ptr<OrderContextManager> ocm_;
  std::unique_ptr<BaseOrderProcessor> processor_;
  std::unique_ptr<BaseOrderGateway> gateway_;
  std::unique_ptr<order_util::OrderLogger> logger_;
  std::unique_ptr<order_util::OrderLogger> order_packet_logger_;
  std::unique_ptr<BaseParser> parser_;
  typename SystemCriticalError::Callback system_critical_error_callback_;

  std::unique_ptr<base::api_base::NativeHttpAsioDriver> http_driver_;
  std::unique_ptr<WsSubscriber> ws_subscriber_;
  std::unique_ptr<RestSubscriber> rest_subscriber_;
  std::unique_ptr<NativePrivateClient> rest_client_;
  std::unique_ptr<IFixTradeClient> fix_client_;
  std::unique_ptr<GrpcTradeClient> grpc_client_;
  GetProductHolderFunc get_product_holder_;

  std::function<void(RequestStatus)> reporter_func_;
  mutable bool account_is_ready_ = false;
};

using IOrderSubsystem = BaseOrderSubsystem;

}  // namespace coin2::exchange::base::order
